package com.bill99.rpc.reference;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.ClassUtils;

import com.bill99.rpc.RpcException;
import com.bill99.rpc.application.ApplicationBean;
import com.bill99.rpc.cluster.LoadBalance;
import com.bill99.rpc.cluster.ServiceNode;
import com.bill99.rpc.cluster.loadbalance.RoundRobinLoadBalance;
import com.bill99.rpc.common.Constants;
import com.bill99.rpc.common.util.NetUtils;
import com.bill99.rpc.config.ReferenceConfig;
import com.bill99.rpc.reference.cache.Cache;
import com.bill99.rpc.reference.cache.CacheManager;
import com.bill99.rpc.registry.NotifyListener;
import com.bill99.rpc.registry.Registry;
import com.bill99.rpc.registry.RegistryBean;
import com.bill99.rpc.registry.url.ReferenceUrl;
import com.bill99.rpc.registry.url.RegistryUrl;
import com.bill99.rpc.registry.url.ServiceUrl;
import com.bill99.rpc.registry.zookeeper.ZookeeperRegistryFactory;
import com.bill99.rpc.registry.zookeeper.client.zkclient.ZkclientZookeeperTransporter;

/**
 * @author sheng.zhang
 * 
 */
public abstract class ReferenceSupport extends ReferenceConfig implements
		BeanClassLoaderAware, FactoryBean, InitializingBean,
		ApplicationContextAware, MethodInterceptor {

	private static final long serialVersionUID = 1L;
	protected final Logger logger = LoggerFactory.getLogger(getClass());
	protected ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
	protected ApplicationContext applicationContext = null;
	protected Object serviceProxy;
	protected List<ServiceNode> serviceNodes;
	protected ReferenceUrl referenceUrl;
	protected CacheManager cacheManager = new CacheManager();
	protected List<InvokeMethod> invokeMethods = new ArrayList<InvokeMethod>();
	private Map<String, LoadBalance> loadBalances = new ConcurrentHashMap<String, LoadBalance>();

	@Override
	public void setBeanClassLoader(ClassLoader classLoader) {
		this.beanClassLoader = classLoader;
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext)
			throws BeansException {
		this.applicationContext = applicationContext;
	}

	protected ClassLoader overrideThreadContextClassLoader() {
		return ClassUtils.overrideThreadContextClassLoader(beanClassLoader);
	}

	protected void resetThreadContextClassLoader(ClassLoader original) {
		if (original != null) {
			Thread.currentThread().setContextClassLoader(original);
		}
	}

	@Override
	public Object invoke(MethodInvocation invocation) throws Throwable {
		ClassLoader originalClassLoader = overrideThreadContextClassLoader();
		try {
			checkReferenceUrl();
			checkServiceNodes();
			LoadBalance loadBalance = getReferenceLoadBalance(referenceUrl);
			boolean failover = isMethodFailover(invocation.getMethod()
					.getName());
			return doInvoke(invocation, loadBalance, failover);
		} finally {
			resetThreadContextClassLoader(originalClassLoader);
		}
	}

	@Override
	public void afterPropertiesSet() {
		this.serviceProxy = new ProxyFactory(getServiceInterface(), this)
				.getProxy(beanClassLoader);
		//
		serviceNodes = new ArrayList<ServiceNode>();
		Map<?, ?> applicationBeans = applicationContext
				.getBeansOfType(ApplicationBean.class);
		Entry<?, ?> applicationEntry = applicationBeans.entrySet().iterator()
				.next();
		ApplicationBean applicationBean = (ApplicationBean) applicationEntry
				.getValue();
		// 设置应用参数信息
		referenceUrl = getReferenceUrl(applicationBean);
		//
		Map<?, ?> map = applicationContext.getBeansOfType(RegistryBean.class);
		// 如果没有配置注册中心，那么直接从配置中读取serviceUrl列表
		if (map.size() > 1)
			throw new RpcException(RpcException.UNKNOWN_EXCEPTION,
					"Please check registry configuration,must have only one registry");
		Entry<?, ?> entry = map.entrySet().iterator().next();
		RegistryBean registryBean = (RegistryBean) entry.getValue();
		String addr = registryBean.getAddress();
		RegistryUrl registryUrl = new RegistryUrl(addr);
		ZookeeperRegistryFactory factory = new ZookeeperRegistryFactory();
		factory.setZookeeperTransporter(new ZkclientZookeeperTransporter());
		//
		cacheManager.addParameter("cacheDir", registryBean.getCacheDir());
		//
		try {
			Registry registry = factory.getRegistry(registryUrl);
			// 查找服务
			List<ServiceUrl> serviceUrls = registry.lookup(referenceUrl);
			for (ServiceUrl url : serviceUrls) {
				ServiceNode node = new ServiceNode(url);
				serviceNodes.add(node);
			}
			// 写入本地缓存
			writeServiceNodesToCache(referenceUrl, serviceUrls);
			// 订阅服务
			registry.subscribe(referenceUrl, new NotifyListener() {
				@Override
				public void notify(List<ServiceUrl> serviceUrls) {
					logger.info(
							"Subscribe service has changed, reference url: [{}]",
							referenceUrl);
					serviceNodes.clear();
					for (ServiceUrl serviceUrl : serviceUrls) {
						ServiceNode node = new ServiceNode(serviceUrl);
						serviceNodes.add(node);
					}
					// 写入本地缓存
					writeServiceNodesToCache(referenceUrl, serviceUrls);
				}
			});
		} catch (Throwable t) {
			logger.error(t.getMessage(), t);
			readServiceNodesFromCache(referenceUrl);
		}
	}

	/**
	 * 将ServiceUrl写入到本地缓存中
	 * 
	 * @param referenceUrl
	 * @param serviceUrls
	 */
	protected void writeServiceNodesToCache(ReferenceUrl referenceUrl,
			List<ServiceUrl> serviceUrls) {
		logger.info("write services to local file,referenceUrl: [{}]",
				referenceUrl);
		Cache cache = cacheManager.getCache();
		cache.put(referenceUrl, serviceUrls);
	}

	/**
	 * 从本地文件中读取referenceurl对应的serviceurl
	 * 
	 * @param referenceUrl
	 */
	protected void readServiceNodesFromCache(ReferenceUrl referenceUrl) {
		logger.info("read services from local file,referenceUrl: [{}]",
				referenceUrl);
		Cache cache = cacheManager.getCache();
		List<ServiceUrl> serviceUrls = cache.get(referenceUrl);
		for (ServiceUrl serviceUrl : serviceUrls) {
			ServiceNode node = new ServiceNode(serviceUrl);
			serviceNodes.add(node);
		}
	}

	protected ReferenceUrl getReferenceUrl(ApplicationBean applicationBean) {
		Map<String, String> parameters = new HashMap<String, String>();
		parameters.put(Constants.INTERFACE_KEY, this.getServiceInterface()
				.getName());
		parameters.put(Constants.APPLICATION_KEY, applicationBean.getName());
		return new ReferenceUrl(Constants.HESSIAN_PROTOCOL, NetUtils.getIp(),
				0, null, parameters);
	}

	/**
	 * 将不可用的ServiceNode从当前ServiceNodes中排除
	 * 
	 * @param unavailableServiceNode
	 */
	protected void notifyUnavailableServiceNode(
			ServiceNode unavailableServiceNode) {
		for (ServiceNode node : serviceNodes) {
			if (node.getServiceUrl().equals(
					unavailableServiceNode.getServiceUrl())) {
				node.setAvailable(false);
			}
		}
	}

	protected List<ServiceNode> getAvailableServiceNodes() {
		List<ServiceNode> avaliableServiceNodes = new ArrayList<ServiceNode>();
		for (ServiceNode node : serviceNodes) {
			if (node.isAvailable())
				avaliableServiceNodes.add(node);
		}
		return avaliableServiceNodes;
	}

	@Override
	public Object getObject() throws Exception {
		return serviceProxy;
	}

	@Override
	public Class<?> getObjectType() {
		return getServiceInterface();
	}

	@Override
	public boolean isSingleton() {
		return true;
	}

	public List<InvokeMethod> getInvokeMethods() {
		return invokeMethods;
	}

	public void setInvokeMethods(List<InvokeMethod> invokeMethods) {
		this.invokeMethods = invokeMethods;
	}

	public boolean isMethodFailover(String methodName) {
		for (InvokeMethod invokeMethod : invokeMethods) {
			if (invokeMethod.getName().equals(methodName)) {
				return invokeMethod.isFailover();
			}
		}
		return failover;
	}

	public LoadBalance getReferenceLoadBalance(ReferenceUrl referenceUrl) {
		LoadBalance loadBalance = loadBalances.get(referenceUrl.toFullString());
		if (loadBalance == null) {
			loadBalance = new RoundRobinLoadBalance();
			loadBalances.put(referenceUrl.toFullString(), loadBalance);
		}
		return loadBalance;
	}

	protected void checkReferenceUrl() {
		if (referenceUrl == null) {
			throw new RpcException(RpcException.FAILOVER_EXCEPTION,
					"referenceUrl == null");
		}
	}

	protected void checkServiceNodes() {
		if (serviceNodes == null || serviceNodes.isEmpty()) {
			throw new RpcException(RpcException.FAILOVER_EXCEPTION,
					"Failed to find any service");
		}
	}

	protected abstract Object doInvoke(MethodInvocation invocation,
			LoadBalance loadBalance, boolean failover) throws RpcException;

}
